/***************************************************************************
 *
 * Copyright (c) 2012 Baidu.com, Inc. All Rights Reserved
 * common.h,v 1.0 2012-09-01 15:58 yelu01
 *
 **************************************************************************/



/**
 * @file disk_matrix.cpp
 * @author yelu01(yelu01@baidu.com)
 * @date 2012-09-01 15:58
 * @version 1.0
 * @brief matrix operation on disk using sorted string table.
 *
 **/

#include "disk_matrix.h"

const uint32_t data_block_t::magic = 0x41414141;

data_block_t::data_block_t()
{
	bytes_cnt = 0;
	LOG(DEBUG, "data_block_t constructed.");
}

data_block_t::~data_block_t()
{
	bytes_cnt = 0;
	LOG(DEBUG, "data_block_t deconstructed.");
}

uint64_t data_block_t::serialize(FILE* fp)
{
	// write magic
	uint64_t total_len = 0;
	uint64_t len = fwrite(&magic, sizeof(magic), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize magic.");
	total_len += (sizeof(magic) * len);
	// write item_cnt
	uint64_t item_cnt = items.size();
	len = fwrite(&item_cnt, sizeof(item_cnt), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize item_cnt.");
	total_len += (sizeof(item_cnt) * len);
	//write items
	for(uint64_t i = 0; i < item_cnt; i++)
	{
		len = items[i]->serialize(fp);
		CHECK_TRUE(len > 0, return 0, "failed to serialize item[%d]", i);
		total_len += len;
	}
	return total_len;
}

uint64_t data_block_t::deserialize(FILE* fp)
{
	// read and check magic
	uint64_t total_len = 0;
	uint32_t n = 0;
	uint64_t len = fread(&n, sizeof(magic), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to deserialize magic.");
	CHECK_TRUE(n == magic, return 0, "failed to check magic.[magic:%u]"
			"[n:%u]", magic, n);
	total_len += (sizeof(magic) * len);
	// read item_cnt
	uint64_t item_cnt = 0;
	len = fread(&item_cnt, sizeof(item_cnt), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to deserialize item_cnt.");
	total_len += (sizeof(item_cnt) * len);
	//read items
	// reserve enough space for items to avoid exceptions when push_back.
	try
	{
		items.reserve(item_cnt);
	}
	catch(std::exception& e)
	{
		LOG(WARNING, "err when reserve space for items: %s", e.what());
		return 0;
	}
	for(uint64_t i = 0; i < item_cnt; i++)
	{
		pitem p(new(std::nothrow) item_t);
		CHECK_TRUE(p, return 0, "failed to new memory for items[%d]", i);
		len = p->deserialize(fp);
		CHECK_TRUE(len > 0, return 0, "failed to deserialize item[%d]", i);
		this->add_item(p);
		total_len += len;
	}
	// bytes_cnt is the bytes deserilized.
	bytes_cnt = total_len;
	return total_len;
}

uint64_t data_block_t::bytes()
{
	return bytes_cnt;
}

int data_block_t::add_item(const pitem& item)
{
	items.push_back(item);
	return 0;
}

uint64_t index_t::serialize(FILE* fp)
{
	uint64_t total_len = 0;
	// write last_key.
	uint64_t len = last_key.serialize(fp);
	CHECK_TRUE(len > 0, return 0, "failed to serialize last_key.");
	total_len += len;
	// write offset.
	len = fwrite(&offset, sizeof(offset), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize offset.");
	total_len += (sizeof(offset) * len);
	return total_len;
}

uint64_t index_t::deserialize(FILE* fp)
{
	uint64_t total_len = 0;
	// read last_key.
	uint64_t len = last_key.deserialize(fp);
	CHECK_TRUE(len > 0, return 0, "failed to deserialize last_key.");
	total_len += len;
	// read offset.
	len = fread(&offset, sizeof(offset), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to deserialize offset.");
	total_len += (sizeof(offset) * len);
	return total_len;
}

const uint32_t index_block_t::magic = 0x42424242;

index_block_t::index_block_t()
{
	//LOG(DEBUG, "index_block_t constructed.");
}

index_block_t::~index_block_t()
{
	//LOG(DEBUG, "index_block_t deconstructed.");
}

int index_block_t::add_index(const pindex& index)
{
	indexes.push_back(index);
	return 0;
}

uint64_t index_block_t::serialize(FILE* fp)
{
	uint64_t total_len = 0;
	// write magic
	uint64_t len = fwrite(&magic, sizeof(magic), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize magic.");
	total_len += (sizeof(magic) * len);
	// write index_cnt
	uint64_t index_cnt = indexes.size();
	len = fwrite(&index_cnt, sizeof(index_cnt), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize index_cnt.");
	total_len += (sizeof(index_cnt) * len);
	//write indexes
	for(uint64_t i = 0; i < index_cnt; i++)
	{
		len = indexes[i]->serialize(fp);
		CHECK_TRUE(len > 0, return 0, "failed to serialize indexes[%d]", i);
		total_len += len;
	}
	return total_len;
}

uint64_t index_block_t::deserialize(FILE* fp)
{
	uint64_t total_len = 0;
	// read and check magic
	uint32_t n = 0;
	uint64_t len = fread(&n, sizeof(magic), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to deserialize magic.");
	CHECK_TRUE(n == magic, return 0, "failed to check magic.[magic:%u]"
			"[n:%u]", magic, n);
	total_len += (sizeof(magic) * len);
	// read index_cnt
	uint64_t index_cnt = 0;
	len = fread(&index_cnt, sizeof(index_cnt), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to deserialize index_cnt.");
	total_len += (sizeof(index_cnt) * len);

	// read items
	// reserve enough space for indexes to avoid exceptions when push_back.
	try
	{
		indexes.reserve(index_cnt);
	}
	catch(std::exception& e)
	{
		LOG(WARNING, "err when reserve space for indexes: %s", e.what());
		return 0;
	}
	for(uint64_t i = 0; i < index_cnt; i++)
	{
		pindex p(new(std::nothrow) index_t);
		CHECK_TRUE(p, return 0, "failed to new memory for indexes[%d]", i);
		len = p->deserialize(fp);
		CHECK_TRUE(len > 0, return 0, "failed to deserialize indexes[%d]", i);
		this->add_index(p);
		total_len += len;
	}
	// size is the bytes deserialized.
	return total_len;
}

const uint32_t summary_t::magic = 0x43434343;

uint64_t summary_t::serialize(FILE* fp)
{

	uint64_t total_len = 0;
	// write magic
	uint64_t len = fwrite(&magic, sizeof(magic), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize magic.");
	total_len += (sizeof(magic) * len);
	// write index_block_offset
	len = fwrite(&index_block_offset, sizeof(index_block_offset), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize index_block_offset.");
	total_len += (sizeof(index_block_offset) * len);
	// write total_item_num
	len = fwrite(&total_item_num, sizeof(total_item_num), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize total_item_num.");
	total_len += (sizeof(total_item_num) * len);
	// write first_key
	len = first_key.serialize(fp);
	CHECK_TRUE(len > 0, return 0, "failed to serialize first_key.");
	total_len += len;
	// write last_key
	len = last_key.serialize(fp);
	CHECK_TRUE(len > 0, return 0, "failed to serialize last_key.");
	total_len += len;
	// write bloom_filter
	len = bloom_filter.serialize(fp);
	CHECK_TRUE(len > 0, return 0, "failed to serialize bloom_filter.");
	total_len += len;
	return total_len;
}

uint64_t summary_t::deserialize(FILE* fp)
{
	uint64_t total_len = 0;
	// read magic
	uint32_t n = 0;
	uint64_t len = fread(&n, sizeof(magic), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to deserialize magic.");
	CHECK_TRUE(n == magic, return 0, "failed to check magic.[magic:%u]"
			"[n:%u]", magic, n);
	// read index_block_offset
	len = fread(&index_block_offset, sizeof(index_block_offset), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize index_block_offset.");
	total_len += (sizeof(index_block_offset) * len);
	// read total_item_num
	len = fread(&total_item_num, sizeof(total_item_num), 1, fp);
	CHECK_TRUE(len == 1, return 0, "failed to serialize total_item_num.");
	total_len += (sizeof(total_item_num) * len);
	// read first_key
	len = first_key.deserialize(fp);
	CHECK_TRUE(len > 0, return 0, "failed to serialize first_key.");
	total_len += len;
	// read last_key
	len = last_key.deserialize(fp);
	CHECK_TRUE(len > 0, return 0, "failed to serialize last_key.");
	total_len += len;
	// read bloom_filter
	len = bloom_filter.deserialize(fp);
	CHECK_TRUE(len > 0, return 0, "failed to serialize bloom_filter.");
	total_len += len;
	return total_len;
}

uint64_t disk_matrix_t::cache_id_seed = 0;

disk_matrix_t::disk_matrix_t(const char* file_path,
		const pcache& cache,
		comparator row_cmp,
		comparator col_cmp)
{
	strncpy(this->file_path, file_path, sizeof(this->file_path));
	this->file_path[sizeof(this->file_path) - 1] = '\0';
	this->cache = cache;
	cache_id = cache_id_seed;
	cache_id_seed++;
	this->row_cmp = row_cmp;
	this->col_cmp = col_cmp;
	LOG(DEBUG, "file_map_t constructed.");
}

disk_matrix_t::~disk_matrix_t()
{
	LOG(DEBUG, "file_map_t deconstructed.");
}

int create()
{
	return 0;
}

int disk_matrix_t::load()
{
	// open file.
	FILE* fp = fopen (file_path, "rb");
	CHECK_NOT_NULL(fp, return -1, "failed to open file.");
	// read summary offset.
	int ret = fseek(fp, 0 - sizeof(summary.summary_offset), SEEK_END);
	CHECK_EQ_ZERO(ret, fclose(fp);return -1, "failed to seek summary_offset");
	uint64_t len = fread(&(summary.summary_offset),
			sizeof(summary.summary_offset), 1, fp);
	CHECK_TRUE(len == 1, fclose(fp);return -1, "failed to fread summary_offset.");
	// load summary.
	ret = fseek(fp, summary.summary_offset, SEEK_SET);
	CHECK_EQ_ZERO(ret, fclose(fp);return -1, "failed to seek summary");
	len = summary.deserialize(fp);
	CHECK_TRUE(len > 0, fclose(fp);return -1, "failed to deserilize summary.");
	// load index_block.
	ret = fseek(fp, summary.index_block_offset, SEEK_SET);
	CHECK_EQ_ZERO(ret, fclose(fp);return -1, "failed to seek index_block");
	len = index_block.deserialize(fp);
	CHECK_TRUE(len > 0, fclose(fp);return -1, "failed to deserilize index_block.");

	fclose(fp);
	return 0;
}

int disk_matrix_t::find(const bytes_t& row, std::vector<pitem>& res)
{
	// Find offset in index block.
	uint64_t offset = 0;
	int ret = _bsearch_index_block(index_block, row, offset);
	if(0 != ret) { return -1; }
	// gen cache id.
	// The first 24 bits for cache id which is unique to every file_map, which
	// corresponding to 1677,7216 files.
	// The following 40 bits for offset, which corresponding to 1T file.
	uint64_t id = cache_id << 40;
	memcpy(&id, &offset, 5);
	// Get data_block form cache.
	pdata_block p;
	if(cache)
	{
		p = cache->find<data_block_t>(id);
	}
	// If not found or no cache. Read data block and find it in data block.
	if(!p)
	{
		p.reset(new(std::nothrow) data_block_t);
		CHECK_TRUE(p, return -1, "failed to alloc mem for data_block_t.");
		FILE* fp = fopen(file_path, "rb");
		ret = fseek(fp, offset, SEEK_SET);
		CHECK_EQ_ZERO(ret, fclose(fp);return -1, "failed to seek offset.");
		uint64_t len = p->deserialize(fp);
		CHECK_TRUE(len > 0, fclose(fp);return -1,"failed to deserialize data_block.");
		fclose(fp);
		// If there is cache. Insert it into cache.
		if(cache)
		{
			cache->insert(id, p);
		}
	}
	// find in data_block.
	ret = _bsearch_data_block(p, row, res);
	return ret;
}

int disk_matrix_t::_bsearch_index_block(const index_block_t& index_block,
		 const bytes_t& target, uint64_t& offset)
{
	const std::vector<pindex>& indexes = index_block.indexes;
	if(indexes.size() <= 0) { return -1; }
	// If target < summary.first_key or target > summary.last_key,
	// We will fail to find it.
	if(row_cmp(target, summary.first_key) || row_cmp(summary.last_key, target))
	{
		return -1;
	}
	uint64_t low = 0, high = indexes.size() - 1;
	uint64_t mid = 0;
	while(low < high)
	{
		mid = low + (high - low)/2;		// avoid overflow.
		// indexes[mid] == target, then return.
		if(!row_cmp(indexes[mid]->last_key, target)){ high = mid; }
		else {low = mid + 1;}
	}

	// If reaching here, low is the first which is bigger than or equal to target.
	// This is quite wired, take a snack then beat your brain. :(
	offset = indexes[low]->offset;
	return 0;
}

int disk_matrix_t::_bsearch_data_block(const pdata_block& data_block,
	 	 	const bytes_t& target, std::vector<pitem>& res)
{
	if(!data_block)
	{
		LOG(WARNING, "data_block is null ptr.");
		return -1;
	}
	const std::vector<pitem>& items = data_block->items;
	uint64_t low = 0, high = items.size() - 1;
	uint64_t mid = 0, curr_pos = UINT64_MAX;
	while(low < high)
	{
		mid = low + (high - low)/2;
		if(row_cmp(items[mid]->row, target)){ low = mid + 1; }
		else if(row_cmp(target, items[mid]->row)){ high = mid - 1; }
		else
		{
			if(mid < curr_pos) { curr_pos = mid; }
			high = mid;
		}
	}
	if(curr_pos == UINT64_MAX) { return -1; }
	// copy matched items to res, the reason why doing a copy is the data_block
	// maybe destroyed by cache, so hold a reference to item in data_block.items
	// is not safe.
	row_eq_cmp index_eq(row_cmp);
	do
	{
		res.push_back(items[curr_pos]);
		curr_pos++;
	}while(curr_pos < items.size() &&
			index_eq(items[curr_pos]->row, items[curr_pos - 1]->row));
	return 0;
}

int disk_matrix_t::save(iterator_t& ite)
{
	// Note: the size can only be used to create bloom filter, because when doing
	// merge, the iterator will not read all items to memory, thus input items
	// cannot give exact size, the size given is the sum of all items pending
	// for merging.
	uint64_t size = ite.size();
	CHECK_TRUE(size > 0, return -1, "ite is empty.");
	// sort the items.
	// std::sort(items.begin(), items.end(), vec_cmp(row_cmp, col_cmp));
	FILE* fp = NULL;
	uint64_t curr_size = 0;
	pdata_block curr_data_block;			//Maybe big, user heap.
	pindex curr_index;						// Small, use stack.
	row_ne_cmp index_ne(row_cmp);
	bytes_t pre_key;
	uint64_t i = 0;
	ite.reset();
	pitem curr_item;
	while(ite.get_next(curr_item) == 0)
	{
		// If this is the first item, do some initilization work.
		// Why not removing it outside the loop? We cannot know whether there is
		// items unless we enter the loop.
		if(0 == i)
		{
			// open file.
			fp = fopen(file_path, "wb");
			CHECK_NOT_NULL(fp, return -1, "failed to open file.");
			// write data block. a data block is at least data_block_size bytes.
			int ret = summary.bloom_filter.create(size, 0.01);
			CHECK_EQ_ZERO(ret, fclose(fp);return -1, "failed to create bloom filter.");
			summary.first_key = curr_item->row;
			pre_key = curr_item->row;
			curr_data_block.reset(new(std::nothrow) data_block_t);
			CHECK_TRUE(curr_data_block, fclose(fp);return -1,
						"failed to alloc memory for curr_data_block.");
			curr_index.reset(new(std::nothrow) index_t);
			CHECK_TRUE(curr_index, fclose(fp);return -1,
									"failed to alloc memory for curr_index.");
			curr_index->offset = ftell(fp);
		}
		// Check if we need to output a data block.
		// If the following conditions happen, output it:
		// 1.curr_size exceed data_block_size
		// 2.the curr key is not equal to pre_key
		// 3.this is not the first key.
		if(0 != i && curr_size >= data_block_size &&
				index_ne(curr_item->row, pre_key))
		{
			curr_index->last_key = pre_key;
			index_block.add_index(curr_index);
			uint64_t len = curr_data_block->serialize(fp);
			CHECK_TRUE(len > 0, fclose(fp);return -1,
					"failed to save curr_data_block.");
			// A new block will begin. Refresh state.
			curr_size = 0;
			curr_data_block.reset(new(std::nothrow) data_block_t);
			CHECK_TRUE(curr_data_block, fclose(fp);return -1,
						"failed to alloc memory for curr_data_block.");
			curr_index.reset(new(std::nothrow) index_t);
			CHECK_TRUE(curr_index, fclose(fp);return -1,
									"failed to alloc memory for curr_index.");
			curr_index->offset = ftell(fp);
		}
		// Begin to write item like normal.
		// Copy item to summary.last_key, add item to data_block, change size,
		// add current to bloom filter.
		summary.last_key = curr_item->row;
		curr_data_block->add_item(curr_item);
		curr_size += curr_item->size();
		summary.bloom_filter.insert((void*)(curr_item->row), curr_item->row.size());
		i++;
		pre_key = curr_item->row;
	}
	// Write the remaining items.
	if(curr_data_block->items.size() > 0)
	{
		curr_index->last_key = pre_key;
		index_block.add_index(curr_index);
		uint64_t len = curr_data_block->serialize(fp);
		CHECK_TRUE(len > 0, fclose(fp);return -1,"failed to save curr_data_block.");
	}

	// write index_block to file and reg info to summary.
	summary.total_item_num = i;
	summary.index_block_offset = ftell(fp);
	uint64_t len = index_block.serialize(fp);
	CHECK_TRUE(len > 0, fclose(fp);return -1, "failed to save index_block.");
	// write summary to file.
	summary.summary_offset = ftell(fp);
	len = summary.serialize(fp);
	CHECK_TRUE(len > 0, fclose(fp);return -1, "failed to save summary.");
	// write summary offset to file.
	len = fwrite(&(summary.summary_offset), sizeof(summary.summary_offset), 1, fp);
	CHECK_TRUE(1 == len, fclose(fp);return -1, "fwrite summary.summary_offset to file failed.");

	fclose(fp);
	return 0;
}

disk_matrix_iterator_t::disk_matrix_iterator_t(disk_matrix_t& disk_matrix):
		_disk_matrix(disk_matrix),
		_index_block(_disk_matrix.index_block)
{
	_index_in_data_block = 0;
	_index_in_index_block = 0;
	_is_reset = false;
}

int disk_matrix_iterator_t::reset()
{
	_data_block.reset();
	const index_block_t& index_block = _disk_matrix.index_block;
	if(index_block.indexes.size() <= 0)
	{
		return -1;
	}
	const char* file_path = _disk_matrix.file_path;
	_index_in_index_block = 0;
	// read the first data block;
	pdata_block p(new(std::nothrow) data_block_t);
	CHECK_TRUE(p, return -1, "failed to alloc mem for "
			"data_block_t.");
	// open file.
	FILE* fp = fopen (file_path, "rb");
	CHECK_NOT_NULL(fp, return -1, "failed to open file.");
	// load data_block.
	int ret = fseek(fp, index_block.indexes[0]->offset, SEEK_SET);
	CHECK_EQ_ZERO(ret, fclose(fp);return -1,
			"failed to seek data_block.");
	uint64_t len = p->deserialize(fp);
	CHECK_TRUE(len > 0, fclose(fp);return -1,
			"failed to deserilize summary.");
	if(p->items.size() <= 0)
	{
		return -1;
	}
	_data_block = p;
	_index_in_data_block = 0;
	fclose(fp);
	_is_reset = true;
	return 0;
}

int disk_matrix_iterator_t::get_next(pitem& item)
{
	if(!_data_block) { return -1; }
	if(_is_reset)
	{
		item = _data_block->items[_index_in_data_block];
		_is_reset = false;
		return 0;
	}
	// If current data block has not been read over, continue to iterate.
	if(_index_in_data_block < _data_block->items.size() - 1)
	{
		_index_in_data_block++;
		item = _data_block->items[_index_in_data_block];
		return 0;
	}

	// Reaching here means a new data block. Read the next data block;
	if(_index_in_index_block < _index_block.indexes.size() - 1)
	{
		_index_in_index_block++;
		_index_in_data_block = 0;
		_data_block.reset(new(std::nothrow) data_block_t);
		CHECK_TRUE(_data_block, return -1, "failed to alloc mem for data_block_t.");
		// open file.
		const char* file_path = _disk_matrix.file_path;
		FILE* fp = fopen(file_path, "rb");
		CHECK_NOT_NULL(fp, return -1, "failed to open file.");
		// load data_block.
		int ret = fseek(fp, _index_block.indexes[_index_in_index_block]->offset,
						SEEK_SET);
		CHECK_EQ_ZERO(ret, fclose(fp);return -1, "failed to seek data_block.");
		uint64_t len = _data_block->deserialize(fp);
		CHECK_TRUE(len > 0, fclose(fp);return -1, "failed to deserilize summary.");
		if(_data_block->items.size() <= 0)
		{
			return -1;
		}
		item = _data_block->items[_index_in_data_block];
		return 0;
	}
	// The end
	return -1;
}

uint64_t disk_matrix_iterator_t::size()
{
	return _disk_matrix.summary.total_item_num;
}


